package com.lagou.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.springframework.util.StopWatch;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 每隔5s打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
 * <p>
 * TP概念: 按照请求时间升序排序，取占比下(向上取整)最大的请求时间
 * example(10 requests): [1ms,3ms,4ms,10ms,20ms,30ms,50ms,80ms,100ms,200ms]
 * TP90: 10 * 90 / 100.0 = 9 ==> 100ms
 * TP99: 10 * 99 / 100.0 ~= 10 ===> 200ms
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    private volatile int state = 0;

    private static final Long MONITOR_TIME = 5 * 1000L;
    private static final Long TP_MINUTE_TIME = 1L;
    private static final Long TP_TIME = TP_MINUTE_TIME * 60 * 1000;

    private final Map<String, Performance> sumMap = new ConcurrentHashMap<String, Performance>();

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String isAsync = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY);
        if (Boolean.parseBoolean(isAsync)) {
            // 异步调用不加入统计
            return invoker.invoke(invocation);
        }
        StopWatch watch = new StopWatch();
        try {
            watch.start();
            return invoker.invoke(invocation);
        } finally {
            watch.stop();
            long lastTaskMs = watch.getLastTaskTimeMillis();
            String serviceName = invocation.getServiceName();
            String methodName = invocation.getMethodName();
            String key = serviceName + "|" + methodName;
            sumMap.putIfAbsent(key, new Performance());
            sumMap.get(key).addRequestTime(lastTaskMs);
            startMonitor();
        }
    }

    private void startMonitor() {
        if (state == 0) {
            synchronized (sumMap) {
                if (state == 0) {
                    Timer timer = new Timer();
                    timer.scheduleAtFixedRate(new MonitorTask(), MONITOR_TIME, MONITOR_TIME);
                    state = 1;
                }
            }
        }
    }

    private class MonitorTask extends TimerTask {

        public void run() {
            sumMap.forEach((key, performance) -> {
                performance.validRequest();
                if (performance.totalRequest > 0) {
                    String[] split = key.split("\\|");
                    String serviceName = split[0];
                    String methodName = split[1];
                    System.out.println(String.format("%tF %<tT %s Method[%s] %d Minute TP Info: ", new Date(), serviceName, methodName, TP_MINUTE_TIME));
                    performance.printTp(new TPUnit(90, 100));
                    performance.printTp(new TPUnit(99, 100));
                }
            });
        }
    }

    private static class TPUnit {
        private int value;
        private int rate;

        public TPUnit(int value, int rate) {
            this.value = value;
            this.rate = rate;
        }

        public int getIndex(Long total) {
            return (int) Math.ceil(total * value * 1.0 / rate);
        }
    }

    private static class Performance {

        private Long totalRequest = 0L;
        private List<RequestInfo> requestList = new ArrayList<>();

        public synchronized void addRequestTime(Long costTime) {
            totalRequest += 1;
            requestList.add(new RequestInfo(costTime));
        }

        public void printTp(TPUnit tpUnit) {
            List<RequestInfo> list;
            Long total;
            synchronized (this) {
                list = new ArrayList<>(requestList);
                total = totalRequest;
            }

            list.sort(Comparator.comparingLong(RequestInfo::getCostTime));
            RequestInfo info = list.get(tpUnit.getIndex(total - 1));
            System.out.println(String.format("TP%d TOTAL[%d]: %sms.", tpUnit.value, total, info.costTime));
        }

        public void validRequest() {
            final long now = System.currentTimeMillis();
            synchronized (this) {
                int beforeSize = requestList.size();
                requestList = requestList.stream().filter(v -> (now - v.requestStamp) <= TP_TIME).collect(Collectors.toList());
                totalRequest = totalRequest - (beforeSize - requestList.size());
            }
        }

        private static class RequestInfo {
            private Long costTime;
            private Long requestStamp;

            public RequestInfo(Long costTime) {
                this.costTime = costTime;
                this.requestStamp = System.currentTimeMillis();
            }

            public Long getCostTime() {
                return costTime;
            }

            public void setCostTime(Long costTime) {
                this.costTime = costTime;
            }

            public Long getRequestStamp() {
                return requestStamp;
            }

            public void setRequestStamp(Long requestStamp) {
                this.requestStamp = requestStamp;
            }
        }
    }

}
